package com.changhong.framework.sdfs.namenode.server.register;

import com.changhong.framework.sdfs.namenode.server.builder.NameNodeBuilder;
import com.changhong.framework.sdfs.namenode.server.config.NameNodeConfig;
import com.changhong.framework.sdfs.namenode.server.datanode.DataNodeInfo;
import com.changhong.framework.sdfs.namenode.server.datanode.DataNodeLease;
import com.changhong.framework.sdfs.namenode.server.utils.DataNodeIdUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


/**
 * 数据节点注册中心
 * @author summer
 */
public class DefaultDataNodeRegistry implements DataNodeRegistry{
    private static final Logger logger = LoggerFactory.getLogger(DefaultDataNodeRegistry.class);
    private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private final Lock read = readWriteLock.readLock();
    private final Lock write = readWriteLock.writeLock();
    private final ConcurrentHashMap<String, Map<String, DataNodeLease<DataNodeInfo>>> registry
            = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduledExecutorService= Executors.newScheduledThreadPool(1);
    private final NameNodeConfig nameNodeConfig;
    private volatile Boolean checkDataNodeIsRunning=true;



    public DefaultDataNodeRegistry(){
        this.nameNodeConfig= NameNodeBuilder.getInstance(NameNodeConfig.class);
        scheduledExecutorService.scheduleAtFixedRate(new MonitorCheckDataNodeIsAlive(),nameNodeConfig.getCheckHeartbeatIntervalSeconds()
        ,nameNodeConfig.getCheckHeartbeatIntervalSeconds(), TimeUnit.SECONDS);
    }

    @Override
    public void register(DataNodeInfo dataNode) {
        read.lock();
        try {
            Map<String, DataNodeLease<DataNodeInfo>> gMap = registry.get(dataNode.getAppName());
            if (gMap == null) {
                final ConcurrentHashMap<String, DataNodeLease<DataNodeInfo>> gNewMap = new ConcurrentHashMap<String, DataNodeLease<DataNodeInfo>>();
                gMap = registry.putIfAbsent(dataNode.getAppName(), gNewMap);
                if (gMap == null) {
                    gMap = gNewMap;
                }
            }
            synchronized (this) {
                //查找是否有注册的数据节点对象
                DataNodeLease<DataNodeInfo> existingLease = gMap.get(DataNodeIdUtils.generalDataNodeId(dataNode));
                if (existingLease == null) {
                    existingLease = new DataNodeLease<>(dataNode);
                    gMap.put(DataNodeIdUtils.generalDataNodeId(dataNode), existingLease);
                }
                //如果已经存在了 说明可能存在重复的注册情况 可以把续约的时间更新了
                existingLease.renew();
            }

        } finally {
            read.unlock();
        }
    }

    public void shutDownCheckRenew(){
        checkDataNodeIsRunning=false;
    }

    private Map<String, DataNodeLease<DataNodeInfo>> getDataNodeLease(String appName){
        return registry.get(appName);
    }

    @Override
    public boolean cancel(String appName, String id) {
        if (getDataNodeLease(appName) != null) {
            getDataNodeLease(appName).remove(id);
            return false;
        }
        logger.warn("不存在的数据节点下线 appName:{}和id:{}",appName,id);
        return true;
    }

    @Override
    public boolean renew(String appName, String id) {
        logger.debug("续约的数据节点名称:{},实例id:{}",appName,id);
        if (getDataNodeLease(appName) != null) {
            getDataNodeLease(appName).get(id).renew();
            return true;
        }
        logger.warn("不存在的数据节点心跳续约 appName:{}和id:{}",appName,id);
        return false;
    }

    @Override
    public void evict() {
        //TODO 暂时不做
    }

    public void remove(String appName, String id){
        getDataNodeLease(appName).remove(id);
    }



    class MonitorCheckDataNodeIsAlive implements Runnable{
        @Override
        public void run() {
            try {
                logger.debug("心跳检查任务调度开始执行......");
                registry.entrySet().forEach(element->{
                    String appName = element.getKey();
                   Collection<DataNodeLease<DataNodeInfo>> dataNodeLeases = element.getValue().values();
                   for(DataNodeLease<DataNodeInfo> dataNodeInfoDataNodeLease: dataNodeLeases){
                       if(System.currentTimeMillis()-dataNodeInfoDataNodeLease.getLastRenewalTimestamp()>nameNodeConfig.getDataNodeRenewMs()){
                           logger.debug("发现超时的数据节点:{},当前的续约时间:{}",appName,dataNodeInfoDataNodeLease.getLastRenewalTimestamp());
                           remove(appName,DataNodeIdUtils.generalDataNodeId(dataNodeInfoDataNodeLease.getHolder()));
                       }
                   }
                });
                logger.debug("心跳检查任务调度结束执行......");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
